package com.xdja.kafka.hdfs.sink;



import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.sink.SinkConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 向hdfs写入的connector
 */
public class HdfsSinkConnector extends SinkConnector {
    private static final Logger log = LoggerFactory.getLogger(SinkConnector.class);
    public static final String HDFS_URL = "hdfs.url";
    public static final String HDFS_PATH = "hdfs.path";
    private static final ConfigDef CONFIG_DEF = new ConfigDef()
            .define(HDFS_URL, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "hdfs url")
            .define(HDFS_PATH, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "hdfs path");
    private String hdfsUrl;
    private String hdfsPath;

    @Override
    public String version() {
        String version="1.0.0";
        return version;
    }
    @Override
    public void start(Map<String, String> props) {
        log.info("开始执行stop()方法");
        System.out.println("开始执行HdfsSinkConnector的start()方法");
        hdfsUrl = props.get(HDFS_URL);
        hdfsPath = props.get(HDFS_PATH);
        log.info("获取到的hdfsUrl:{}, hdfsPath:{}", hdfsUrl, hdfsPath);
    }

    @Override
    public Class<HdfsSinkTask> taskClass() {
        log.info("开始执行taskClass()方法");
        System.out.println("开始执行HdfsSinkConnector的taskClass()方法");
        return HdfsSinkTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        log.info("开始执行taskConfigs()方法");
        System.out.println("开始执行HdfsSinkConnector的taskConfigs()方法");
        List<Map<String, String>> configs = new ArrayList<Map<String, String>>();
        for (int i = 0; i < maxTasks; i++) {
            Map<String, String> config = new HashMap<String, String>();
            if (hdfsUrl != null) {
                config.put(HDFS_URL, hdfsUrl);
            }
            if (hdfsPath != null) {
                config.put(HDFS_PATH, hdfsPath);
            }
            configs.add(config);
        }
        return configs;
    }

    @Override
    public void stop() {
        log.info("开始执行stop()方法");
        System.out.println("开始执行HdfsSinkConnector的stop()方法");
    }

    @Override
    public ConfigDef config() {
        log.info("开始执行config()方法");
        System.out.println("开始执行HdfsSinkConnector的config()方法");
        return CONFIG_DEF;
    }
}
